package com.shujia.spark

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo19Partition {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()

    //会在yarn中显示的名称
    conf.setAppName("wc")

    /**
      * spark shuffle之后rdd默认的分区数据
      *
      */
    conf.set("spark.default.parallelism", "3")

    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    /**
      * 第一个rdd的分区数据默认等于切片数，可以指定一个最小分区数，最小分区所默认是2
      *
      */
    val linesRDD: RDD[String] = sc.textFile("spark/data/words", 10)


    println("linesRDD:" + linesRDD.getNumPartitions)

    /**
      * 没有shuffle算子----->后面的rdd分区数默认等于父rdd分区数
      *
      */

    val wordRDD: RDD[String] = linesRDD.flatMap(_.split(","))

    println("wordRDD:" + wordRDD.getNumPartitions)

    val kvRDD: RDD[(String, Int)] = wordRDD.map((_, 1))


    println("kvRDD:" + kvRDD.getNumPartitions)

    /**
      *
      * countRDD 分区数决定因素
      *
      * 1、如果没有指定分区数据默认等于父rdd分区数
      * 2、shuffle类的算子可以手动指定分区数据（reduce的数量）
      *
      *
      * 分区数据决定优先级： 手动指定---> 默认分区数（spark.default.parallelism）---> 父rdd分区数
      *
      */

    val countRDD: RDD[(String, Int)] = kvRDD.reduceByKey(_ + _, 2)


    println("countRDD:" + countRDD.getNumPartitions)

    /**
      * repartition: 改变rdd分区数据，会产生shuffle,  不做任何逻辑处理
      *
      * 可以用于提供并行度
      *
      */


    val repartitionRDD: RDD[(String, Int)] = countRDD.repartition(10)


    println("repartitionRDD:" + repartitionRDD.getNumPartitions)

    repartitionRDD.foreach(println)

    while (true) {

    }
  }

}
